package cn.dglydrpy.study.ssm.common.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import cn.dglydrpy.study.ssm.common.kafka.domain.Customer;
/**
 * Kafka生产者
 * @author dglydrpy
 
1.在pom中引入kafka-client的依赖
2. 设置属性
	2.1) bootstrap.servers
	2.2) key.serializer
	2.3) value.serializer
3.获取Producer,将属性设置给Producer
4.声明ProducerRecord,里面需要指定topic,key,value
5.发送消息
	5.1) 同步: producer.send().get()
		5.1.1) 可重试错误(连接中断的重试等)
		5.1.2) 不可重试错误(消息过长等)
	5.2) 异步,Kafka的生产者通过回调的形式,支持对异步发送消息时产生的异常处理
 */
public class MyProducer {

    private Properties properties = new Properties();

    MyProducer(){
        //org.apache.kafka.common.serialization.Serializer
        this.properties.put("bootstrap.servers","192.168.80.128:9092");
        this.properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
      //this.properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        //自定义序列化规则
        this.properties.put("value.serializer","cn.dglydrpy.study.ssm.common.kafka.serializer.CustomerSerializer");
    }

//    public Producer<String,String> getProducer(){
//        return new KafkaProducer<String, String>(this.properties);
//    }
//   
//    public void sendMessage(String topic,String key,String value){
//        try{
//            ProducerRecord<String,String> record = new ProducerRecord<>(topic,key,value);
//            //同步
//            RecordMetadata recordMetadata = getProducer().send(record).get();
//            //getProducer().send(record,new MyProducerCallback());
//        }catch (Exception e){
//            e.printStackTrace();
//        }
//    }

    public static void main(String[] args) {
        MyProducer myProducer = new MyProducer();
        //myProducer.sendMessage("myTopic","key1","Hello World");
        Customer customer = new Customer();
        customer.setId(1);
        customer.setCustomerName("李华");
        myProducer.sendMessage("myTopic","key2",customer);
    }

    public Producer<String, Customer> getProducer(){
        return new KafkaProducer<String, Customer>(this.properties);
    }
    
	private void sendMessage(String topic, String key, Customer customer) {
		try {
			ProducerRecord<String, Customer> record = new ProducerRecord<>(topic, key, customer);
			RecordMetadata recordMetadata = getProducer().send(record).get();
			// getProducer().send(record,new MyProducerCallback());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
//异步
/*class MyProducerCallback implements Callback {

    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {

    }
}*/
